package com.jiang.starter.redis.channel;

import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.lang.reflect.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;

/**
 * 动态代理类
 */
public class ChannelInvocationHandler implements InvocationHandler {

    /**
     * 构造并返回代理对象
     */
    public static Object generateProxyClass(Class<?> type, ApplicationContext applicationContext) {
        return Proxy.newProxyInstance(type.getClassLoader(), new Class[]{type}, new ChannelInvocationHandler(type, applicationContext));
    }

    private Class<?> type;
    private ApplicationContext applicationContext;
    private Class<?> messageType;
    private String topicName;
    private RedisTemplate<String, Object> redisTemplate;
    private RedisMessageListenerContainer redisMessageListenerContainer;

    /**
     * 由于是工厂创建，所以需要亲自注入参数
     */
    public ChannelInvocationHandler(Class<?> type, ApplicationContext applicationContext) {
        this.type = type;
        this.applicationContext = applicationContext;

        // 获取泛型信息
        Type[] genericInterfaces = type.getGenericInterfaces();
        // 遍历该类型的所有接口
        for (Type genericInterface : genericInterfaces) {
            if (genericInterface instanceof ParameterizedType) {
                ParameterizedType parameterizedType = (ParameterizedType) genericInterface;
                // 如果是Channel接口
                if(Channel.class.isAssignableFrom((Class<?>) parameterizedType.getRawType())) {
                    // 获取泛型类型
                    Type actualTypeArgument = parameterizedType.getActualTypeArguments()[0];
                    if(messageType == null) {
                        messageType = (Class<?>) actualTypeArgument;
                    } else {
                        throw new RuntimeException("一个Channel只能继承一个Channel接口！");
                    }
                }
            }
        }

        // 获取Topic信息
        if(type.isAnnotationPresent(Topic.class)) {
            topicName = type.getAnnotation(Topic.class).value();
        }

        // 获取RedisTemplate对象
        redisTemplate = applicationContext.getBean("redisTemplate", RedisTemplate.class);

        // 获取RedisMessageListenerContainer对象
        redisMessageListenerContainer = applicationContext.getBean(RedisMessageListenerContainer.class);

        // 添加监听器
        addReceiveListener();
    }

    /**
     * 动态代理方法入口
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws InvocationTargetException, IllegalAccessException, InterruptedException {

        // 如果不是Channel接口，则不代理
        if(proxy instanceof Channel == false) {
            return method.invoke(args);
        }

        // 代理publish方法
        if ("publish".equals(method.getName())) {

            // 获取消息对象
            if (args.length != 1) {
                throw new RuntimeException("发布消息方法一次只能传入一个参数");
            }
            Object arg = args[0];

            // 发布
            redisTemplate.convertAndSend(topicName, arg);
        }

        // 代理subscribe方法
        else if("subscribe".equals(method.getName())) {

            // 获取消费者
            if (args.length != 1) {
                throw new RuntimeException("订阅方法一次只能传入一个参数");
            }
            Object arg = args[0];
            Consumer consumer = (Consumer) arg;

            // 将消费者添加到消费者集合中
            consumers.add(consumer);
        }

        // 代理receive方法
        else if("receive".equals(method.getName())) {

            // 创建CompletableFuture用于异步获取数据
            CompletableFuture<Object> future = new CompletableFuture<>();

            // 将future对象放入队列中等待异步被取出执行
            blockingQueue.put(future);

            // 阻塞直到有数据返回
            return future.join();
        }

        return null;
    }

    /**
     * 初始化时添加监听器
     * 1、执行消费者集合中所有消费方法
     * 2、给阻塞的接收线程返回数据
     */
    private List<Consumer> consumers = Collections.synchronizedList(new ArrayList<>());
    private BlockingQueue<CompletableFuture<Object>> blockingQueue = new LinkedBlockingQueue<>();
    private void addReceiveListener() {
        redisMessageListenerContainer.addMessageListener((message, pattern) -> {

            // 获取消息
            byte[] bytes = message.getBody();

            // 1、执行消费者集合中所有消费方法
            for (Consumer consumer : consumers) {

                // 反序列化消息
                Object data = redisTemplate.getValueSerializer().deserialize(bytes);

                // 执行消费方法
                consumer.accept(data);
            }

            // 2、为每一个future都创建一份数据返回
            int N = blockingQueue.size();
            for (int i = 0; i < N; i++) {
                CompletableFuture<Object> future = blockingQueue.poll();

                // 反序列化消息
                Object data = redisTemplate.getValueSerializer().deserialize(bytes);

                // 将对象通过future返回
                future.complete(data);
            }
        }, ChannelTopic.of(topicName));
    }

}
